package com.bstek.bdf4.mq.core;

import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONUtil;
import com.bstek.bdf2.core.mq.publisher.RabbitMQRequestEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.util.Date;



/**
 * @Title: 消息发送确认
 * ConfirmCallback  只确认消息是否正确到达 Exchange 中
 * ReturnCallback   消息没有正确到达队列时触发回调，如果正确到达队列不执行
 * 1. 如果消息没有到exchange,则confirm回调,ack=false
 * 2. 如果消息到达exchange,则confirm回调,ack=true
 * 3. exchange到queue成功,则不回调return
 * 4. exchange到queue失败,则回调return

 * @Package: com.bstek.bdf4.mq.core
 * @Description:
 * @Author: zxq
 * @Date: 2024/2/6 - 16:38
 */
@Slf4j
@Component
public class RabbitProducerAckConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    @Autowired
    private RabbitConfig rabbitConfig;

    /**
     *
     * 通过实现 MessagePostProcessor 接口来对消息进行额外的处理，例如修改消息的属性、添加额外的标记等。
     *  配置统一head...等
     * @param
     * @return org.springframework.amqp.core.MessagePostProcessor
     */
    public static final MessagePostProcessor messagePostProcessor =
            message -> {
                // 对消息进行后处理操作,可以修改消息的属性、添加额外的标记等
                message.getMessageProperties().setHeader("content_type", MessageProperties.CONTENT_TYPE_JSON);
                message.getMessageProperties().setType("bstek");
                return message;
            };


    /**
     * ==================设置请求body数据结构为：json=====================
     */
    @Bean
    public RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(rabbitConfig.producerJackson2MessageConverter());
        //异步回调，定义ConfirmCallback，MQ返回结果时会回调这个ConfirmCallback
        rabbitTemplate.setConfirmCallback(this);
        //发送消息时，如果消息无法路由到任何队列，就会触发 ReturnCallback 的回调方法，将未能路由的消息返回给发送者。
        rabbitTemplate.setReturnCallback(this);
        //定义消息路由失败时的策略。true，则调用ReturnCallback；false：则直接丢弃消息
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }

    /**
     * ==================配置admin管理MQ相关信息=====================
     * 通过使用 RabbitAdmin，我们可以在应用程序中方便地进行 RabbitMQ 的管理操作，如声明和删除交换机、队列和绑定等。
     * 可以通过调用 RabbitAdmin 的相应方法来执行这些操作，例如 declareExchange、declareQueue、declareBinding、deleteExchange、deleteQueue 等。
     */
    @Bean
    public RabbitAdmin rabbitAdmin(final ConnectionFactory createConnectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(createConnectionFactory);
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    /**
     * confirm机制只保证消息到达exchange，不保证消息可以路由到正确的queue,如果exchange错误，就会触发confirm机制
     *
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 消息发送时间
        Date date = new Date();
        String format = DateUtil.format(date, "yyyy-MM-dd HH:mm:ss");
        String id = correlationData.getId();
        String jsonData = new String(correlationData.getReturnedMessage().getBody());
        RabbitMQRequestEntity rabbitMQRequest = JSONUtil.toBean(JSONUtil.toJsonStr(jsonData), RabbitMQRequestEntity.class);
        if (ack) {
            log.info("[MQProducerAckConfig.confirm]消息发送成功通道id[{}]时间[{}]  BODY={}", id, format, rabbitMQRequest);
        } else {
            log.error("[MQProducerAckConfig.confirm]消息发送失败通道id[{}]时间[{}] cause={}  BODY={}", id, format, cause, rabbitMQRequest);
        }
    }

    /**
     * Return 消息机制用于处理一个不可路由的消息。在某些情况下，如果我们在发送消息的时候，当前的 exchange
     * 不存在或者指定路由 key 路由不到，这个时候我们需要监听这种不可达的消息就需要这种return机制
     *
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        RabbitMQRequestEntity rabbitMQRequest = JSONUtil.toBean(JSONUtil.toJsonStr(new String(message.getBody())), RabbitMQRequestEntity.class);
        String format = DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss");
        // 反序列化对象输出
        log.info("[ReturnCallback.returnedMessage调用]消息送达MQ异常_业务id[{}]，时间[{}]  消息主体: {}  应答码: {} 描述: {} 消息使用的交换器: {}  消息使用的路由键: {}"
                , rabbitMQRequest.getBusinessId()
                , format
                , rabbitMQRequest.getData()
                , replyCode
                , replyText
                , exchange
                , routingKey);
    }




}
